package exams.concurrent;

import java.util.Random;
import java.util.concurrent.*;

public class PubSubSolution {

    public final static Object lock = new Object();

    double a = 0.0000000000000000000000000000000000000000e1000000000;

    public static LinkedBlockingDeque<String> queue = new LinkedBlockingDeque<>();

    public final static Random random = new Random();

    public static void main(String[] args) {
        ExecutorService executorService1 = Executors.newFixedThreadPool(2);
        ExecutorService executorService2 = Executors.newFixedThreadPool(2);

        for(int i =0 ; i < 5000; i++) {
            executorService1.submit(new Producer("Msg_" + i));
            executorService2.submit(new Consumer());
        }

        executorService1.shutdown();
        executorService2.shutdown();
    }

    static class Producer implements Runnable {

        private final String msg;

        public Producer(String msg) {
            this.msg = msg;
        }

        @Override
        public void run() {
            synchronized (lock) {
                while (queue.size() == 100) {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                mockSpentTime();
                queue.add(this.msg);
                System.out.println("Producer [" + Thread.currentThread().getName() + "] send msg: " + this.msg);
                lock.notifyAll();
            }
        }
    }

    static class Consumer implements Runnable {

        public Consumer() {

        }

        @Override
        public void run() {
            synchronized (lock) {
                while (queue.isEmpty()) {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                mockSpentTime();
                String msg = queue.poll();
                System.out.println("Consumer [" + Thread.currentThread().getName() + "] received msg: " + msg);
                lock.notifyAll();
            }
        }
    }

    public static void mockSpentTime() {
        try {
            Thread.sleep((random.nextInt() % 4 + 1) * 100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
